package edu.purdue.cs.aggr;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import edu.purdue.cs.aggr.AccumulatorFactory.AccumulatorType;
import edu.purdue.cs.aggr.message.VertexMoveCommand;
import edu.purdue.cs.aggr.message.VertexSetupMessage;

public class Controller extends Node {

	public static List<String> nodes = new ArrayList<String>();
	public static Map<String, String> location = new HashMap<String, String>();
	public static Map<String, List<String>> children = new HashMap<String, List<String>>();
	public static List<VertexData> leaves = new ArrayList<VertexData>();
	public static List<VertexData> nonLeaves = new ArrayList<VertexData>();
	public static int updatesExpected = 0;
	public static int combinesExpected = 0;
  public static int combinesReceived = 0;
	public static long combineStarted = 0;
	
	public Controller(String IP, int port) {
		super(IP, port, true);
	}

	public Map<String, VertexSetupMessage> vertexConfigs = new HashMap<String, VertexSetupMessage>();
	
	@Override
	public void run() {
//		System.out.println("Starting controller ... ");
//		System.out.println("Found " + nodes.size() + " nodes...");
//		System.out.println(nodes);
//		boolean returnnow = true;
//		if (returnnow)
//			return;
		
		try {
			Thread.sleep(10000); // A small delay so that all vertices start
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		
		if (Main.experiment.equals("topk")) {
			//TOPK
			synchronized (vertexConfigs) {
				leaves = new ArrayList<VertexData>();
				nonLeaves = new ArrayList<VertexData>();
				
				System.out.println("fanout :" + Main.fanout);
				System.out.println("leaves :" + Main.leaves);

				chooseSetup();
 				executeTopKExperiment(leaves, nonLeaves);
			}
		} else if (Main.experiment.equals("sort")) {
				//SORT
			synchronized (vertexConfigs) {
				leaves = new ArrayList<VertexData>();
				nonLeaves = new ArrayList<VertexData>();

				System.out.println("fanout :" + Main.fanout);
				System.out.println("leaves :" + Main.leaves);
				
				chooseSetup();
				executeSortExperiment(leaves, nonLeaves);
			}
		} else if (Main.experiment.equals("grep")) {
		//		//GREP
			synchronized (vertexConfigs) {
				leaves = new ArrayList<VertexData>();
				nonLeaves = new ArrayList<VertexData>();

				System.out.println("fanout :" + Main.fanout);
				System.out.println("leaves :" + Main.leaves);
				
				chooseSetup();
				executeGrepExperiment(leaves, nonLeaves);
			}
		} else if (Main.experiment.equals("wordcount")) {
		//		//WORD COUNT
			synchronized (vertexConfigs) {
				leaves = new ArrayList<VertexData>();
				nonLeaves = new ArrayList<VertexData>();

				System.out.println("fanout :" + Main.fanout);
				System.out.println("leaves :" + Main.leaves);
				
				chooseSetup();
				executeWordCountExperiment(leaves, nonLeaves);
			}
	    } else if (Main.experiment.equals("y")) {
	    	//Y Experiment
	    	synchronized (vertexConfigs) {
	    		leaves = new ArrayList<VertexData>();
	    		nonLeaves = new ArrayList<VertexData>();

				System.out.println("fanout :" + Main.fanout);
				System.out.println("leaves :" + Main.leaves);
				
				chooseSetup();
	    		executeYExperiment(leaves, nonLeaves);
	    	}
		
	    } else if (Main.experiment.contains("mb") && Main.experiment.length() == 3) {
	    	//microbenchmarks
	    	Integer mbtype = Integer.parseInt(new String() + Main.experiment.charAt(2));
	    	synchronized (vertexConfigs) {
	    		leaves = new ArrayList<VertexData>();
	    		nonLeaves = new ArrayList<VertexData>();

				System.out.println("fanout :" + Main.fanout);
				System.out.println("leaves :" + Main.leaves);
				
				chooseSetup();
	    		executeMicroBenchExperiment(leaves, nonLeaves, mbtype);
	    	}
		
	    } else {
	    	throw new IllegalArgumentException("Unknown experiment");
	    }
		
		while (true) {
			synchronized (vertexConfigs) {
				if (vertexConfigs.size() > 0) {
					for (String destinationId : vertexConfigs.keySet()) {
						VertexSetupMessage setupMessage = vertexConfigs.get(destinationId);
						setupMessage.setDestination(destinationId);
						getSender().sendMessage(setupMessage, destinationId);
						System.out.println("Send setup message with leaf=" + setupMessage.isInputReader() + " to " + destinationId);
					}
				}
				vertexConfigs.clear();
			}
			
			try {
				synchronized (this) {
					wait();
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public void addVertexSetupMessage (String id, VertexSetupMessage setupMessage) {
		synchronized (vertexConfigs) {
			vertexConfigs.put(id, setupMessage);
		}
	}
	
	public void combineNodes(String id) {
	  /* given node n with id=id,
	   * put all nodes in subtree of n in the VertexManager of n */
	  moveChildren(id, children.get(id));
	}
	
	private void moveChildren(String id, List<String> children) {
	  if(children == null) return;
	  for (String child : children) {
	    moveChildren(id, Controller.children.get(child));
	    // send message over the wire to make child go to id
	    sendMoveCommand(child, id);
	    // update the child's real location
	    location.put(child, id);
	  }
	}
	
	private void sendMoveCommand(String id, String target) {
	  combinesExpected++;
	  System.out.println("sending move command: " + id + " target: " + target);
	  /* move node with given id to the destination */
	  VertexMoveCommand message = new VertexMoveCommand();
	  message.setDestination(getLocation(id));
    message.setTargetId(getLocation(target));
	  message.setVertexId(id);
	  this.getSender().sendMessage(message, getLocation(id));
	}
	
	public String getLocation(String id) {
	  return location.get(id);
	}
	
	private void chooseSetup() {

		setupLeaves(leaves, nonLeaves, Main.leaves, Main.fanout);
/*		if (Main.leaves == 16) {
			switch(Main.fanout) {
			case 2:
				setup16LeavesK2(leaves, nonLeaves);
				break;
			case 4:
				setup16LeavesK4(leaves, nonLeaves);
				break;
			case 16:
				setup16LeavesK16(leaves, nonLeaves);
				break;
			default:
				throw new IllegalArgumentException("Fanout value of " + Main.fanout + " is not supported for 16 leaves.");
			}
		} else if (Main.leaves == 64) {
			switch(Main.fanout) {
			case 2:
				setup64LeavesK2(leaves, nonLeaves);
				break;
			case 4:
				setup64LeavesK4(leaves, nonLeaves);
				break;
			case 8:
				setup64LeavesK8(leaves, nonLeaves);
				break;
			case 64:
				setup64LeavesK64(leaves, nonLeaves);
				break;
			default:
				throw new IllegalArgumentException("Fanout value of " + Main.fanout + " is not supported for 64 leaves.");	
			}
		} else {
		   throw new IllegalArgumentException(Main.leaves + " leaves is not supported.");
		}*/
	}
			
	private int numberOfNodesNeeded(Integer leafcount, Integer fanout) {
		int result = leafcount;
		Double levelcount = new Double(Math.ceil(new Double(leafcount)/new Double(fanout)));
		
		if (leafcount == 1) {return 1;}
		
		while (levelcount > 0.) {
			if (levelcount == 1.) {
				result++;
				levelcount = 0.;
			} else {
				result += (int)Math.ceil(levelcount);
				levelcount = new Double(Math.ceil(new Double(levelcount)/new Double(fanout)));
			}
		}
		
		return result;
	}
	
	private void setupLeaves(List<VertexData> leafs, List<VertexData> nonLeafs, Integer leafcount, Integer fanout) {
		// make sure that we have enough nodes to set up the tree
		int nodecount = numberOfNodesNeeded(leafcount, fanout);
		System.out.println("Nodes needed: " + nodecount);
		System.out.println("Nodes available: " + nodes.size());
		if (nodes.size() < nodecount) {
			throw new IllegalArgumentException("Too few nodes in the list (expected " + nodecount + ", got " + nodes.size() + ")");
		}
		
		//the first parent is the first non-leaf node
		int parent = leafcount;
		int nextlevel = parent;
		// set up the leaves
		for (int i=0; i<leafcount; i++) {
			if (i > 0 && i % fanout==0) {
				parent++;
			}
			leafs.add(new VertexData(true, getPort(), getPort(), getPort(), nodes.get(i), nodes.get(parent)));
			System.out.println("Leaf created at " + nodes.get(i));
		}
		System.out.println(leafcount +" children created.");
		
		// determine the number of nodes in this level
		int numinlevel = (int)Math.ceil(new Double(leafcount)/new Double(fanout));
		// initialize the child counter. This is incremented in nonLeafs.add.
		int child =-1;

		while (numinlevel > 0) {
			System.out.println("Creating " + numinlevel + " in next level");
			int start = nextlevel;
			nextlevel = parent + 1;
			parent = nextlevel;
			for (int i = 0; i + start < nextlevel; i++) {
				if (i > 0 && i % fanout==0) {
					parent++;
				}
				
				// create a list with all nodes which are children of this one
				LinkedList <String> children = new LinkedList<String>();
				for (int x = 0; x < fanout && child < start; x++) {
					children.add(nodes.get(++child));
				}
				
				// store children of this node in map
				this.children.put(nodes.get(i+start), children);
				
				// determine if this is the root and create the vertex appropriately
				if (parent < nodecount) {
					System.out.println("Creating middle node at " + nodes.get(i+start));
					nonLeafs.add(new VertexData(false, getPort(), getPort(), getPort(), nodes.get(i+start), nodes.get(parent),children));
				} else {
					System.out.println("Creating root at " + nodes.get(i + start));
					nonLeafs.add(new VertexData(false, getPort(), getPort(), getPort(), nodes.get(i+start), "",children));
				}
			}
			if (numinlevel == 1) {
				numinlevel = 0;
			} else{
				numinlevel = (int)Math.ceil(new Double(numinlevel)/new Double(fanout));
			}
		}
	}
		
	private void executeTopKExperiment(List<VertexData> leafs, List<VertexData> nonLeafs) {
		System.out.println("Executing topK experiment...");
		for (VertexData data : leafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.TOPK);
			msg1.setInputReader(true);
			msg1.setDestinationVertex(data.destination);
			addVertexSetupMessage(data.id, msg1);
		}
		
		for (VertexData data : nonLeafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.TOPK);
			msg1.setInputReader(false);
			msg1.setDestinationVertex(data.destination);
//			System.out.println("Sources vertex count of " + data.id + " is " + data.sources.size());
			msg1.setSourceVertices(data.sources);
			addVertexSetupMessage(data.id, msg1);
		}
	}
	
	private void executeSortExperiment(List<VertexData> leafs, List<VertexData> nonLeafs) {
		for (VertexData data : leafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.SORT);
			msg1.setInputReader(true);
			msg1.setDestinationVertex(data.destination);
			addVertexSetupMessage(data.id, msg1);
		}
		
		for (VertexData data : nonLeafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.SORT);
			msg1.setInputReader(false);
			msg1.setDestinationVertex(data.destination);
//			System.out.println("Sources vertex count of " + data.id + " is " + data.sources.size());
			msg1.setSourceVertices(data.sources);
			addVertexSetupMessage(data.id, msg1);
		}
	}
	
	private void executeGrepExperiment(List<VertexData> leafs, List<VertexData> nonLeafs) {
		System.out.println("Executing grep experiment...");
		for (VertexData data : leafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.GREP);
			msg1.setInputReader(true);
			msg1.setDestinationVertex(data.destination);
			addVertexSetupMessage(data.id, msg1);
		}
		
		for (VertexData data : nonLeafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.GREP);
			msg1.setInputReader(false);
			msg1.setDestinationVertex(data.destination);
//			System.out.println("Sources vertex count of " + data.id + " is " + data.sources.size());
			msg1.setSourceVertices(data.sources);
			addVertexSetupMessage(data.id, msg1);
		}
	}
	
	private void executeYExperiment(List<VertexData> leafs, List<VertexData> nonLeafs) {
		System.out.println("Executing Y experiment...");
		for (VertexData data : leafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.YExperiment);
			msg1.setInputReader(true);
			msg1.setDestinationVertex(data.destination);
			addVertexSetupMessage(data.id, msg1);
		}
		
		for (VertexData data : nonLeafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.YExperiment);
			msg1.setInputReader(false);
			msg1.setDestinationVertex(data.destination);
//			System.out.println("Sources vertex count of " + data.id + " is " + data.sources.size());
			msg1.setSourceVertices(data.sources);
			addVertexSetupMessage(data.id, msg1);
		}
	}
	
	private void executeMicroBenchExperiment(List<VertexData> leafs, List<VertexData> nonLeafs, Integer growth) {
		System.out.println("Executing MicroBenchmark experiment...");
		for (VertexData data : leafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.MICROBENCH);
			msg1.setAccumulatorSubtype(growth);
			msg1.setInputReader(true);
			msg1.setDestinationVertex(data.destination);
			addVertexSetupMessage(data.id, msg1);
		}
		
		for (VertexData data : nonLeafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.MICROBENCH);
			msg1.setAccumulatorSubtype(growth);
			msg1.setInputReader(false);
			msg1.setDestinationVertex(data.destination);
//			System.out.println("Sources vertex count of " + data.id + " is " + data.sources.size());
			msg1.setSourceVertices(data.sources);
			addVertexSetupMessage(data.id, msg1);
		}
	}
	
	private void executeWordCountExperiment(List<VertexData> leafs, List<VertexData> nonLeafs) {
		System.out.println("Executing wordcount experiment...");
		for (VertexData data : leafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.WORDCOUNT);
			msg1.setInputReader(true);
			msg1.setDestinationVertex(data.destination);
			addVertexSetupMessage(data.id, msg1);
		}
		
		for (VertexData data : nonLeafs) {
			VertexSetupMessage msg1 = new VertexSetupMessage();
			msg1.setAccumulatorType(AccumulatorType.WORDCOUNT);
			msg1.setInputReader(false);
			msg1.setDestinationVertex(data.destination);
//			System.out.println("Sources vertex count of " + data.id + " is " + data.sources.size());
			msg1.setSourceVertices(data.sources);
			addVertexSetupMessage(data.id, msg1);
		}
	}
		
	/**
	 * Just a temporary class to hold input data for a vertex
	 * @author chamikara
	 *
	 */
	public static class VertexData {
		private String id;
		private String destination = null;
		private List<String> sources = new ArrayList<String>();
		private boolean leaf = false;
		private int sourcePort;
		private int destinationPort;
		private int selfPort;
		
		public int getSelfPort() {
			return selfPort;
		}

		/**
		 * data[0] = id;
		 * data[1] = source id
		 * data[rest] = destination ids
		 * @param data
		 */
		public VertexData(boolean leaf, int selfPort,  int sourcePort, int destinationPort, String... data) {
			this.sourcePort = sourcePort;
			this.destinationPort = destinationPort;
			this.selfPort = selfPort;
			this.leaf = leaf;
			id = Vertex.createId( Vertex.extractIP(data[0]), selfPort);
			
			if (data.length > 1) {
			  if (data[1]!=null) {
				destination = Vertex.createId( Vertex.extractIP(data[1]), destinationPort);
			  }
			
			  if (data.length > 2) {
			    for (int i=2; i<data.length; i++) {
				  sources.add(Vertex.createId(Vertex.extractIP(data[i]), sourcePort));
			    }
			  }
			}
		}

		public VertexData(boolean leaf, int selfPort,  int sourcePort, 
				      int destinationPort, String selfdata, String parentdata,
				      LinkedList <String> childdata) {
			this.sourcePort = sourcePort;
			this.destinationPort = destinationPort;
			this.selfPort = selfPort;
			this.leaf = leaf;
			
			id = Vertex.createId( Vertex.extractIP(selfdata), selfPort);
			
			if (parentdata.length() > 1) {
				destination = Vertex.createId( Vertex.extractIP(parentdata), destinationPort);
			}
			
			for (String child : childdata) {
				  sources.add(Vertex.createId(Vertex.extractIP(child), sourcePort));
			}
		}

		public String getId() {
			return id;
		}

		public String getDestination() {
			return destination;
		}

		public List<String> getSources() {
			return sources;
		}

		public boolean isLeaf() {
			return leaf;
		}

		public int getSourcePort() {
			return sourcePort;
		}

		public int getDestinationPort() {
			return destinationPort;
		}

		public String toString() {
			String s = "";
			if (leaf) {
				s = s + "leaf-,";
			} else {
				s = s + "non-leaf-,";
			}
			s = s + id;
			s = s + ", destination-" + destination;
			
			s = s + ", sources-";
			for (String source : sources) {
				s = s + ", " + source;
			}
			return s;
		}
		
	}
	
	public static void main(String[] args) {
		List<VertexData> leafNodes = new ArrayList<Controller.VertexData>();
		List<VertexData> nonLeafNodes = new ArrayList<Controller.VertexData>();
		
		
		Controller c = new Controller("controllerIP", 20000);
		
//		for (int i=0; i<127; i++) {
//			nodes.add("node " + (i + 1));
//		}
//		c.setup64LeavesK2(leafNodes, nonLeafNodes);
		
//		for (int i=0; i<85; i++) {
//			nodes.add("node " + (i + 1));
//		}
//		c.setup64LeavesK4(leafNodes, nonLeafNodes);
		
//		for (int i=0; i<73; i++) {
//			nodes.add("node " + (i + 1));
//		}
//		c.setup64LeavesK8(leafNodes, nonLeafNodes);
		
//		for (int i=0; i<65; i++) {
//			nodes.add("node " + (i + 1));
//		}
//		c.setup64LeavesK64(leafNodes, nonLeafNodes);
//		
//		for (VertexData data : leafNodes) {
//			System.out.println(data);
//		}
//		for (VertexData data : nonLeafNodes) {
//			System.out.println(data);
//		}
	}
	
	public static String getIndexOf(String toFind) {
	  //System.out.println("index: " + toFind + " " + Arrays.deepToString(nodes.toArray()));
	  if(nodes.contains(toFind))
	    return nodes.indexOf(toFind) + "";
	  else
	    return toFind;
	}
}
